// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/array/builder_decimal.h>

#include <cstdint>
#include <memory>

#include <nebula/array/data.h>
#include <nebula/core/buffer.h>
#include <nebula/core/buffer_builder.h>
#include <turbo/utility/status.h>
#include <turbo/base/checked_cast.h>
#include <nebula/numeric/decimal.h>

namespace nebula {

    class Buffer;

    class MemoryPool;

    // ----------------------------------------------------------------------
    // Decimal128Builder

    Decimal128Builder::Decimal128Builder(const std::shared_ptr<DataType> &type,
                                         MemoryPool *pool, int64_t alignment)
            : FixedSizeBinaryBuilder(type, pool, alignment),
              decimal_type_(turbo::checked_pointer_cast<Decimal128Type>(type)) {}

    turbo::Status Decimal128Builder::append(Decimal128 value) {
        TURBO_RETURN_NOT_OK(FixedSizeBinaryBuilder::Reserve(1));
        unsafe_append(value);
        return turbo::OkStatus();
    }

    void Decimal128Builder::unsafe_append(Decimal128 value) {
        value.ToBytes(GetMutableValue(length()));
        byte_builder_.UnsafeAdvance(16);
        unsafe_append_to_bitmap(true);
    }

    void Decimal128Builder::unsafe_append(std::string_view value) {
        FixedSizeBinaryBuilder::unsafe_append(value);
    }

    turbo::Status Decimal128Builder::finish_internal(std::shared_ptr<ArrayData> *out) {
        std::shared_ptr<Buffer> data;
        TURBO_RETURN_NOT_OK(byte_builder_.finish(&data));
        std::shared_ptr<Buffer> null_bitmap;
        TURBO_RETURN_NOT_OK(null_bitmap_builder_.finish(&null_bitmap));

        *out = ArrayData::create(type(), length_, {null_bitmap, data}, null_count_);
        capacity_ = length_ = null_count_ = 0;
        return turbo::OkStatus();
    }

    // ----------------------------------------------------------------------
    // Decimal256Builder

    Decimal256Builder::Decimal256Builder(const std::shared_ptr<DataType> &type,
                                         MemoryPool *pool, int64_t alignment)
            : FixedSizeBinaryBuilder(type, pool, alignment),
              decimal_type_(turbo::checked_pointer_cast<Decimal256Type>(type)) {}

    turbo::Status Decimal256Builder::append(const Decimal256 &value) {
        TURBO_RETURN_NOT_OK(FixedSizeBinaryBuilder::Reserve(1));
        unsafe_append(value);
        return turbo::OkStatus();
    }

    void Decimal256Builder::unsafe_append(const Decimal256 &value) {
        value.ToBytes(GetMutableValue(length()));
        byte_builder_.UnsafeAdvance(32);
        unsafe_append_to_bitmap(true);
    }

    void Decimal256Builder::unsafe_append(std::string_view value) {
        FixedSizeBinaryBuilder::unsafe_append(value);
    }

    turbo::Status Decimal256Builder::finish_internal(std::shared_ptr<ArrayData> *out) {
        std::shared_ptr<Buffer> data;
        TURBO_RETURN_NOT_OK(byte_builder_.finish(&data));
        std::shared_ptr<Buffer> null_bitmap;
        TURBO_RETURN_NOT_OK(null_bitmap_builder_.finish(&null_bitmap));

        *out = ArrayData::create(type(), length_, {null_bitmap, data}, null_count_);
        capacity_ = length_ = null_count_ = 0;
        return turbo::OkStatus();
    }

}  // namespace nebula
